package com.ourhours.redismq;

import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

/**
 * 消息发送
 * 
 */
@Service
public class RedisMQMessageSender {
	Logger logger = LoggerFactory.getLogger(RedisMQMessageSender.class);
	
	@Autowired
	RedisUtil redisUtil;

	@Autowired
	@Qualifier(value = "messageTrunktaskExecutor")
	private ThreadPoolTaskExecutor threadPool;

	/**
	 * 失败重试次数，超过此值则不再重试，默认3次
	 */
	private int failRetryTimes = 3;

	/**
	 * 如果线程池满了，生产者暂停的时间，单位：S
	 */
	private int threadPoolFullSleepSeconds = 1;

	public ThreadPoolTaskExecutor getThreadPool() {
		return threadPool;
	}

	public void setThreadPool(ThreadPoolTaskExecutor threadPool) {
		this.threadPool = threadPool;
	}

	public int getFailRetryTimes() {
		return failRetryTimes;
	}

	public void setFailRetryTimes(int failRetryTimes) {
		this.failRetryTimes = failRetryTimes;
	}

	public int getThreadPoolFullSleepSeconds() {
		return threadPoolFullSleepSeconds;
	}

	public void setThreadPoolFullSleepSeconds(int threadPoolFullSleepSeconds) {
		this.threadPoolFullSleepSeconds = threadPoolFullSleepSeconds;
	}

	/**
	 * 推送消息
	 * 
	 * @param message
	 * 
	 * @return null表示失败
	 */
	public Long put(final Message message) {
		String queue = MtConstant.KEY_PREFIX + message.getDestination();
		// 添加到队列名称集合
		redisUtil.sadd(MtConstant.QUEUE_NAME_SET, queue);
		//增加进队数量
		redisUtil.hincrby(MtConstant.QUEUE_IN_NUMBERS, queue, 1L);
		// 发送消息
		return redisUtil.rpush(queue, message, 0);
	}

	/**
	 * 推送消息
	 * 
	 * @param message
	 * 
	 * @return null表示失败
	 */
	public boolean groupput(List<Message> messageList) {
		List<String> keys = new ArrayList<String>();

		Jedis jedis = null;
		try {
			jedis = redisUtil.getConnent();
			//使用pipline
			Pipeline pipeline = jedis.pipelined();
			for (Message message : messageList) {
				String queue = MtConstant.KEY_PREFIX + message.getDestination();
				keys.add(queue);
				// 添加到队列名称集合
				pipeline.sadd(MtConstant.QUEUE_NAME_SET, queue);
				//增加进队数量
				redisUtil.hincrby(MtConstant.QUEUE_IN_NUMBERS, queue, 1L);
				// 发送消息
				byte[] bytes = ConvertUtil.serialize(message);
				pipeline.rpush(queue.getBytes(), bytes);
			}
			//执行pipline
			pipeline.sync();
		} catch (Exception e) {
			logger.error("redis lpush data failed , keys = " + keys.toString(), e);
			return false;
		} finally {
			redisUtil.close(jedis);
		}
		return true;
	}

}
